package com.sunwayworld.basemodule.common.eptstorage.acqconfig.net;

import com.sunwayworld.basemodule.common.eptstorage.acqconfig.bean.LimsEquiptAcqConfigInstBean;
import com.sunwayworld.basemodule.common.eptstorage.util.Util;
import com.sunwayworld.cloud.module.equipt.acqconfig.bean.CoreEquiptAcqConfigBean;
import com.sunwayworld.framework.constant.Constant;
import com.sunwayworld.framework.exception.InvalidDataException;
import com.sunwayworld.framework.utils.StringUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.util.CharsetUtil;

import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 每个UDP设备占用一个本地端口
 * @auther zhangs
 * @description
 * @date 2022/3/4
 */
public class NetEptUdpSender extends NetEptSender{
    public NetEptUdpSender(CoreEquiptAcqConfigBean eptAcqBean) {
        super(eptAcqBean);
    }
    private Channel channel;
    private NioEventLoopGroup group;
    private int localPort;
    private CountDownLatch waitReadLatch = null;
    private CountDownLatch waitConnectLatch = null;
    private String msg = null;
    public String getMsg() {
        return msg;
    }
    public void setMsg(String msg) {
        this.msg = msg;
    }

    @Override
    public synchronized boolean getConnection() {
        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
        String host = eptAcqBean.getIp();
        int port = Integer.parseInt(eptAcqBean.getPort());
        boolean connect = isConnect();
        if(!connect){
            getLogger().info(Thread.currentThread().getName() + "---try getConnection for remoteAddress={}",(host+":"+port));
            close();
            //尝试连接一次
            tryConnect();
        }else{
            getLogger().info(Thread.currentThread().getName() + "---reuse getConnection for remoteAddress={}",(host+":"+port));
        }
        return isConnect();
    }

    private void tryConnect() {
        group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST,true)
                    .handler(new SimpleChannelInboundHandler<DatagramPacket>() {
                        @Override
                        protected void channelRead0(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket) {
                            String body = datagramPacket.content().toString(CharsetUtil.UTF_8);
                            String threadName = Thread.currentThread().getName();
                            getLogger().info(threadName+"----from remote={},messageReceived={}",datagramPacket.sender(),body);
                            setMsg(body);
                            if(waitReadLatch!=null){
                                waitReadLatch.countDown();
                            }
                        }

                        @Override
                        public void channelActive(ChannelHandlerContext ctx) {
                            setConnect(true);
                            getLogger().info(Thread.currentThread().getName() + "---connectSuccess--------->remoteAddress={}", ctx.channel().remoteAddress());
                            cacheSender();
                            waitConnectLatch.countDown();
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                            super.channelInactive(ctx);
                            getLogger().error("-----channelInactive--------->");
                            close();
                        }

                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                            super.exceptionCaught(ctx, cause);
                            getLogger().error("-----exceptionCaught--------->cause={}",cause);
                        }

                        @Override
                        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception{
                            super.channelUnregistered(ctx);
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(0);
            channelFuture.await(waitConnectSecond, TimeUnit.SECONDS);
            channel = channelFuture.channel();
            setConnect(true);
            String threadName = Thread.currentThread().getName();
            InetSocketAddress localAddress = (InetSocketAddress) channel.localAddress();
            getLogger().info(threadName+"--{},-bindSuccess--------->localAddress={}",getKey(getEptAcqBean()),localAddress);
            cacheSender();
            String key = getEptAcqBean().getIp()+":"+getEptAcqBean().getPort();
            //将配置缓存到内存中
            if(!NetEptSenderHelper.cacheEptAcqConfigMap.containsKey(key)){
                NetEptSenderHelper.cacheEptAcqConfigMap.put(key,getEptAcqBean());
            }

            //连接成功后，线程顺序读取队列中的指令来发送
            startAsyncSequenceReadQueueAndSendWhenActive();

            localPort = localAddress.getPort();
        }catch (Throwable s){
            setConnect(false);
            String threadName = Thread.currentThread().getName();
            getLogger().error(threadName+"-----bind--------->error={}",s);
        }finally {
//            close();
        }
    }

    @Override
    public String send(LimsEquiptAcqConfigInstBean instBean) {
        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
        String host = eptAcqBean.getIp();
        int port = Integer.parseInt(eptAcqBean.getPort());

        if(instBean==null || !isConnect()){
            getLogger().error(Thread.currentThread().getName()+"---sendFail--------->remoteAddress={},cause={}",host+":"+port,"未连接");
            close();
            throw new InvalidDataException("连接已断开，发送指令失败");
        }

        getLogger().info("---------sendMsg={},to={}",instBean.getInstruction(),(host+":"+port));
        ByteBuf byteBuf;
        if (StringUtils.equalsIgnoreCase(Constant.YES, instBean.getHixFlag())) {
            byte[] bytes = Util.hixStringToByteArray(instBean.getInstruction());
            byteBuf = Util.convert2Buffer(bytes);
        } else {
            //字符串命令都要结束符 一般为\n
            String msg = instBean.getInstruction()+System.getProperty("line.separator");
            byteBuf = Util.convert2Buffer(msg);
        }
        channel.writeAndFlush(new DatagramPacket(byteBuf,new InetSocketAddress(host,port)));

        //最多等待2s,如果正常读了，则返回true
        try {
            waitReadLatch = new CountDownLatch(1);
            long t1 = System.currentTimeMillis();
            boolean await = waitReadLatch.await(readWaitSecond, TimeUnit.SECONDS);
            long t2 = System.currentTimeMillis();
            getLogger().info("等待读取数据用时：{},{}",(t2-t1),(host+":"+port));
            String result = "";
            if(await){
                getLogger().info("正常读取数据成功：{},{}",msg,(host+":"+port));
                String msg = getMsg();
                if(msg!=null){
                    result = insertMsg(instBean,msg);
                    setMsg(null);
                }
            }else{
                getLogger().info("读取数据超时：{}",(host+":"+port));
                waitReadLatch.countDown();
                result = "fail";
                //未及时返回消息，可能是网络故障，关闭连接
//                close();
                throw new InvalidDataException("读取数据超时:"+host+":"+port);
            }
            return result;
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new InvalidDataException(e.getMessage());
        }
    }
    @Override
    public void close() {
        getLogger().error("-----close---------");
        setConnect(false);
        if(channel!=null){
            channel.close();
        }
        if(group!=null){
            group.shutdownGracefully();
        }
        NetEptSenderHelper.removeNetEptSender(getEptAcqBean());
    }

}
